package com.gmadmin.mq.listener;

import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author zhangliang
 * @version 1.0
 * @date 2022/8/1 10:02
 */
@RocketMQTransactionListener(rocketMQTemplateBeanName="customMessageTemplate")
public class TransactionListenerImpl implements RocketMQLocalTransactionListener {

    private final static Logger LOGGER = LoggerFactory.getLogger(TransactionListenerImpl.class);

    private AtomicInteger transactionIndex = new AtomicInteger(0);

    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();


    /**
     * 执行本地事务
     * @param message
     * @param o
     * @return
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        /**
         * 以下是本地测试事务，根据不同的条件判断是否提交或者返回
         */
        String transId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        LOGGER.info("#### 本地事务执行, msgTransactionId=[{}]",
                transId);
//
//        int value = transactionIndex.getAndIncrement();
//        int status = value % 3;
//        localTrans.put(transId, status);
//        if (status == 0) {
//            // Return local transaction with success(commit), in this case,
//            // this message will not be checked in checkLocalTransaction()
//            System.out.printf("    # COMMIT # Simulating msg %s related local transaction exec succeeded! ### %n", message.getPayload());
//            return RocketMQLocalTransactionState.COMMIT;
//        }
//
//        if (status == 1) {
//            // Return local transaction with failure(rollback) , in this case,
//            // this message will not be checked in checkLocalTransaction()
//            System.out.printf("    # ROLLBACK # Simulating %s related local transaction exec failed! %n", message.getPayload());
//            return RocketMQLocalTransactionState.ROLLBACK;
//        }
//
//        System.out.printf("    # UNKNOW # Simulating %s related local transaction exec UNKNOWN! \n");
        return RocketMQLocalTransactionState.UNKNOWN;
    }

    /**
     * 检查本地事务状态，先执行上面的
     * @param message
     * @return
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        String transId = (String) message.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;
        Integer status = localTrans.get(transId);
//            if (null != status) {
//                switch (status) {
//                    case 0:
//                        retState = RocketMQLocalTransactionState.COMMIT;
//                        break;
//                    case 1:
//                        retState = RocketMQLocalTransactionState.ROLLBACK;
//                        break;
//                    case 2:
//                        retState = RocketMQLocalTransactionState.UNKNOWN;
//                        break;
//                }
//            }
        LOGGER.info("------ !!! 检查本地事务执行一次," +
                        " msgTransactionId=[{}], TransactionState=[{}] status=[{}]",
                transId, retState, status);

        return retState;
    }
}
